package rpc.consumer.commom.initializer;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import rpc.codec.RpcDecoder;
import rpc.codec.RpcEncoder;
import rpc.constants.RpcConstants;
import rpc.consumer.commom.handler.RpcConsumerHandler;
import io.netty.handler.timeout.IdleStateHandler;
import rpc.exception.processor.ExceptionPostProcessor;
import rpc.threadpool.ConcurrentThreadPool;
import rpc.flow.processor.FlowPostProcessor;

import java.util.concurrent.TimeUnit;

/**
 * @author binghe(公众号：冰河技术)
 * @version 1.0.0
 * @description RpcConsumerInitializer
 */
public class RpcConsumerInitializer extends ChannelInitializer<SocketChannel> {
    private int heartbeatInterval;
    private ConcurrentThreadPool concurrentThreadPool;
    private FlowPostProcessor flowPostProcessor;
    //异常处理后置处理器
    private ExceptionPostProcessor exceptionPostProcessor;


    public RpcConsumerInitializer(int heartbeatInterval, ConcurrentThreadPool concurrentThreadPool, FlowPostProcessor flowPostProcessor, ExceptionPostProcessor exceptionPostProcessor){

        if (heartbeatInterval > 0){
            this.heartbeatInterval = heartbeatInterval;
        }
        this.concurrentThreadPool = concurrentThreadPool;
        this.flowPostProcessor = flowPostProcessor;
        this.exceptionPostProcessor = exceptionPostProcessor;
    }

    @Override
    protected void initChannel(SocketChannel channel) throws Exception {
        ChannelPipeline cp = channel.pipeline();
        cp.addLast(RpcConstants.CODEC_ENCODER, new RpcEncoder(flowPostProcessor));
        cp.addLast(RpcConstants.CODEC_DECODER, new RpcDecoder(flowPostProcessor));
//        cp.addLast(RpcConstants.CODEC_CLIENT_IDLE_HANDLER, new IdleStateHandler(heartbeatInterval, 1000000, 1000000, TimeUnit.MILLISECONDS));
        cp.addLast(RpcConstants.CODEC_HANDLER, new RpcConsumerHandler(concurrentThreadPool, exceptionPostProcessor));
    }

}
